package core;

import bean.RbCmdFactory;
import bean.command.RbCmd;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import handler.BaseHandler;
import handler.exception.DisruptorExceptionHandler;
import lombok.Getter;
import lombok.NonNull;
import lombok.extern.log4j.Log4j2;
import net.openhft.affinity.AffinityStrategies;
import net.openhft.affinity.AffinityThreadFactory;
import org.apache.kafka.clients.producer.Producer;
import producer.ExchangeProducer;
import thirdpart.order.CmdType;
import thirdpart.order.OrderCmd;
import thirdpart.order.OrderDirection;
import thirdpart.order.OrderType;
import thirdpart.uuid.SnowflakeIdWorker;

import java.util.Random;
import java.util.Timer;
import java.util.TimerTask;

@Log4j2
public class EngineCore {

    private final Disruptor<RbCmd> disruptor;

    private static final int RING_BUFFER_SIZE=1024;

    private final Producer producer;

    @Getter
    private final EngineApi api;


    public EngineCore(
            @NonNull final BaseHandler riskHandler,
            @NonNull final BaseHandler matchHandler,
            @NonNull final BaseHandler pubHandler,
            Producer producer)
    {
        this.producer = producer;
        this.disruptor = new Disruptor(
                new RbCmdFactory(),

                RING_BUFFER_SIZE,

                new AffinityThreadFactory("aft_engine_core", AffinityStrategies.ANY),

                ProducerType.SINGLE,

                new BlockingWaitStrategy()
        );

        this.api = new EngineApi(disruptor.getRingBuffer());

    final DisruptorExceptionHandler<RbCmd> exceptionHandler = new DisruptorExceptionHandler<>(
            "main",
            (ex,seq)->{
                log.error("exception thrown on seq={}",seq,ex);
            });
    disruptor.setDefaultExceptionHandler(exceptionHandler);
    // 风控 撮合 发布
    disruptor.handleEventsWith(riskHandler)
    .then(matchHandler)
    .then(pubHandler);
    // 3 启动
        disruptor.start();
        log.info("match engine start");
    // 4.生产线程 定时任务  后续要换成从消息队列读取
        new Timer().schedule(new HqPubTask(123456),1000, 2000);

    }

    private class HqPubTask extends TimerTask{
        private long uid;

        public HqPubTask(long uid) {
            this.uid = uid;
        }

        private SnowflakeIdWorker snowGen1 = new SnowflakeIdWorker(1,1);

        private Random random = new Random();
        /*
        @Override
        public void run() {
            long currentTimeMillis = System.currentTimeMillis();
            api.submitCommand(
                    OrderCmd.builder()
            .code(123456)
            .direction(OrderDirection.BUY)
            .mid((short) uid)
            .oid(snowGen1.nextId())
            .price(random.nextInt(12+1))
            .volume(random.nextInt(12+1))
            .type(CmdType.NEW_ORDER)
            .timestamp(currentTimeMillis)
            .uid(uid)
            .orderType(OrderType.LIMIT)
            .build()
            );
            api.submitCommand(
                    OrderCmd.builder()
                            .type(CmdType.HQ_PUB)
                            .build()
            );

        }

        */
        public void run() {
            long currentTimeMillis = System.currentTimeMillis();
            api.submitCommand( OrderCmd.builder()
                    .code(123456)
                    .direction(OrderDirection.BUY)
                    .mid(uid)
                    .oid(snowGen1.nextId())
                    .price(100)
                    .volume(100)
                    .type(CmdType.NEW_ORDER)
                    .timestamp(currentTimeMillis)
                    .uid(uid)
                    .orderType(OrderType.LIMIT)
                    .build());
            api.submitCommand( OrderCmd.builder()
                    .code(123456)
                    .direction(OrderDirection.SELL)
                    .mid(uid)
                    .oid(snowGen1.nextId())
                    .price(11)
                    .volume(100)
                    .type(CmdType.NEW_ORDER)
                    .timestamp(currentTimeMillis)
                    .uid(uid)
                    .orderType(OrderType.LIMIT)
                    .build());
    }
    }


}
